package com.chukun.flink.stream.operator.base;

import com.chukun.flink.stream.bean.Trade;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.ArrayList;

/**
 * @author chukun
 * @version 1.0.0
 * @description split操作符
 * @createTime 2022年05月10日 00:06:00
 */
public class SplitOperator {

    /**
     * 标记小金额的数据流
     */
    private static final OutputTag<Trade> SMALL_AMOUNT_TAG = new OutputTag<Trade>("small_amount_tag"){};

    /**
     * 标记大金额的数据流
     */
    private static final OutputTag<Trade> LARGE_AMOUNT_TAG = new OutputTag<Trade>("large_amount_tag") {};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Trade> dataStream = env.fromCollection(new ArrayList<Trade>(){
            {
                add(new Trade("12300986754",899,"2018-06"));
                add(new Trade("14500986754",699,"2018-07"));
                add(new Trade("18800986754",88,"2018-08"));
            }
        });

        // 分流操作
        SingleOutputStreamOperator<Trade> processStream = dataStream.process(new ProcessFunction<Trade, Trade>() {
            @Override
            public void processElement(Trade trade, ProcessFunction<Trade, Trade>.Context context, Collector<Trade> collector) throws Exception {
                if (trade.getAmount() < 100) {
                    context.output(SMALL_AMOUNT_TAG, trade);
                } else if (trade.getAmount() > 100) {
                    context.output(LARGE_AMOUNT_TAG, trade);
                }
            }
        });

        DataStream<Trade> smallAmountStream = processStream.getSideOutput(SMALL_AMOUNT_TAG);
        smallAmountStream.print("small amount");


        DataStream<Trade> largeAmountStream = processStream.getSideOutput(LARGE_AMOUNT_TAG);
        largeAmountStream.print("large amount");

        // 触发程序执行
        env.execute("process stream");
    }
}
